package realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.bean.TrafficHomeCourseDetailBean;
import realtime.util.ClickHouseUtil;
import realtime.util.DateFormatUtil;
import realtime.util.MyKafkaUtil;
import realtime.util.MyWindowFunction;

import java.time.Duration;

public class Dws_TrafficHomeCourseDetailPageView {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        /*env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC");
        env.setStateBackend(new HashMapStateBackend());*/

        System.setProperty("HADOOP_USER_NAME", "root");


        DataStreamSource<String> kafkaStream = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", "traffic_home_course_detail_page_view"));

        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    JSONObject jsonObject = JSON.parseObject(value);
                    String pageId = jsonObject.getJSONObject("page").getString("page_id");
                    if ("home".equals(pageId) || "course_detail".equals(pageId)) {
                        out.collect(jsonObject);
                    }
                }
            }
        });

        SingleOutputStreamOperator<JSONObject> jsonStreamWithWatermark = jsonStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                }));

        KeyedStream<JSONObject, String> keyedStream = jsonStreamWithWatermark.keyBy(k -> k.getJSONObject("common").getString("mid"));

        SingleOutputStreamOperator<TrafficHomeCourseDetailBean> javaBeanStream = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, TrafficHomeCourseDetailBean>() {

            private ValueState<String> homeValueState;
            private ValueState<String> courseValueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                StateTtlConfig build = StateTtlConfig.newBuilder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        .build();
                ValueStateDescriptor<String> homeValueStateDescriptor = new ValueStateDescriptor<>("homeValueState", String.class);
                ValueStateDescriptor<String> courseValueStateDescriptor = new ValueStateDescriptor<>("courseValueState", String.class);

                homeValueStateDescriptor.enableTimeToLive(build);
                courseValueStateDescriptor.enableTimeToLive(build);

                homeValueState = getRuntimeContext().getState(homeValueStateDescriptor);
                courseValueState = getRuntimeContext().getState(courseValueStateDescriptor);
            }

            @Override
            public void flatMap(JSONObject value, Collector<TrafficHomeCourseDetailBean> out) throws Exception {
                String homeState = homeValueState.value();
                String courseState = courseValueState.value();
                String dataDate = DateFormatUtil.toDate(value.getLong("ts"));

                long homeCt = 0;
                long courseCt = 0;

                if ("home".equals(value.getJSONObject("page").getString("page_id"))) {
                    if (homeState == null || !homeState.equals(dataDate)) {
                        homeValueState.update(dataDate);
                        homeCt = 1;

                    }
                } else {
                    if (courseState == null || !courseState.equals(dataDate)) {
                        courseValueState.update(dataDate);
                        courseCt = 1;
                    }
                }

                if (homeCt != 0 || courseCt != 0) {
                    out.collect(TrafficHomeCourseDetailBean.builder()
                            .firstCt(homeCt)
                            .courseCt(courseCt)
                            .build());
                }
            }
        });

        SingleOutputStreamOperator<TrafficHomeCourseDetailBean> reduce = javaBeanStream.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficHomeCourseDetailBean>() {
                    @Override
                    public TrafficHomeCourseDetailBean reduce(TrafficHomeCourseDetailBean value1, TrafficHomeCourseDetailBean value2) throws Exception {
                        value1.setFirstCt(value1.getFirstCt() + value2.getFirstCt());
                        value1.setCourseCt(value1.getCourseCt() + value2.getCourseCt());
                        return value1;
                    }
                }, new AllWindowFunction<TrafficHomeCourseDetailBean, TrafficHomeCourseDetailBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TrafficHomeCourseDetailBean> values, Collector<TrafficHomeCourseDetailBean> out) throws Exception {
                        TrafficHomeCourseDetailBean next = values.iterator().next();
                        MyWindowFunction.getWindowFunction(window,next);
                        out.collect(next);
                    }
                });
        reduce.print(">>");
        reduce.addSink(ClickHouseUtil.getSinkFunction("insert into dws_traffic_home_course_detail_view_window values(?,?,?,?,?)"));

        env.execute("Dws_TrafficHomeCourseDetailPageView");


    }
}
